package com.zbj.storm.old;

import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.topology.TopologyBuilder;

/**
 * SimpleTopology
 *
 * @author weigang
 * @create 2019-09-02
 **/
public class SimpleTopology {
    public static void main(String[] args) throws Exception {
        SimpleTopology topology = new SimpleTopology();
        topology.runLocal(10);
    }

    public void runLocal(int waitSeconds) throws Exception {
        TopologyBuilder topologyBuilder = new TopologyBuilder();

        // 第一个参数 Spout的id
        // 第二个参数 要调用的Spout类
        // 第三个参数 节点所需的并行度 可选参数 指示应在集群中执行该组件的线程数 如省略它 Storm则默认为1
        topologyBuilder.setSpout("words", new TestWordSpout(), 1);

        // 只是要通过 shuffleGrouping 指定数据来源"words"
        // shuffle grouping 意味着元组应该从输入任务随机分配到bolt的任务中
        topologyBuilder.setBolt("DoubleAndTripleBolt1", new ExclamationBolt(), 1)
                .shuffleGrouping("words");

        // 一个Bolt可以接收多个数据来源 是要多次调用 shuffleGrouping 即可
        topologyBuilder.setBolt("DoubleAndTripleBolt2", new ExclamationBolt(), 1)
                .shuffleGrouping("words");

        // local测试
        Config config = new Config();
        // 建议加上这行，使得每个bolt/spout的并发度都为1
        config.put(Config.TOPOLOGY_MAX_TASK_PARALLELISM, 1);
        LocalCluster cluster = new LocalCluster();
        cluster.submitTopology("word_count", config, topologyBuilder.createTopology());

        Thread.sleep(waitSeconds * 1000);
        cluster.killTopology("world_count");
        cluster.shutdown();
    }
}